package com.gateway.app.lb;

import com.alibaba.fastjson2.JSON;
import com.gateway.app.common.GatewayUtil;
import com.gateway.app.common.RspEntity;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Publisher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.cloud.gateway.route.Route;
import org.springframework.cloud.gateway.support.ServerWebExchangeUtils;
import org.springframework.core.Ordered;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.http.server.reactive.ServerHttpResponseDecorator;
import org.springframework.stereotype.Component;
import org.springframework.util.StopWatch;
import org.springframework.util.StringUtils;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.URI;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.*;

/**
 * gateway全局拦截
 */
@SuppressWarnings("ALL")
@Slf4j
@Component
public class GatewayFilter implements GlobalFilter, Ordered {
    @Autowired
    private RedisTemplate redisTemplate;

    public static Map<String,Long> KEYS = new HashMap<>();

    @SneakyThrows
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        String method = request.getMethodValue();
        String url = request.getPath().value();
        String sessionId = request.getHeaders().getFirst("Token");
        if (sessionId == null) {
            sessionId = (String) exchange.getAttribute("SID");
        }

        Route route = exchange.getAttribute(ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR);
        if(route != null){
            log.info("route id:{}", route.getId());
        }else {
            log.error("非法请求！");
            return forbidden(exchange);
        }

        String routeId = route.getId();
        // 记录 请求 超过则 提示错误 请求进入时
        //获取一个 可以执行请求的 ip
        List<String> ips =  redisTemplate.opsForList().range(routeId, 0, -1);;
        String taskName = obtainRandomIp(ips,routeId);// obtainRandomIp(ips,routeId,true);
        if(StringUtils.isEmpty(taskName)){
            String ip = GatewayUtil.getIpAddr(request);
            log.error("===========taskName==forbidden black IP:{},URL:{}", ip, url);
            return forbidden(exchange);
        }
        //动态修改路由
        if (route != null) {
            URI uri = URI.create("http://"+taskName);
            Route newRoute = Route.async()
                    .asyncPredicate(route.getPredicate())
                    .filters(route.getFilters())
                    .id(route.getId())
                    .order(route.getOrder())
                    .uri(uri).build();
            //修改路由
            exchange.getAttributes().put(ServerWebExchangeUtils.GATEWAY_ROUTE_ATTR,newRoute);
        }

        StopWatch timer = new StopWatch();
        timer.start(taskName);
        /*return chain.filter(exchange).doFinally(signalType->{
            timer.stop();
            double seconds = timer.getTotalTimeSeconds();
            //请求完成后 remove 回收此 ip
            String key = routeId +"_"+timer.getLastTaskName();
            Long value = redisTemplate.opsForValue().decrement(key);
            log.info("end request taskName:{},seconds:{},key:{},value:{}",timer.getLastTaskName(),timer.getTotalTimeSeconds(),key,value);
            //TODO 回调 通知前端已经请求已经处理完成
            log.info("---signalType:{}", JSON.toJSONString(signalType));
            log.info("end --> {}",timer.prettyPrint());
        });*/
        ServerHttpResponseDecorator decoratedResponse = writeRsp(exchange,routeId,url);

        return chain.filter(exchange.mutate().response(decoratedResponse).build()).doFinally(signalType->{
            timer.stop();
            double seconds = timer.getTotalTimeSeconds();
            //请求完成后 remove 回收此 ip
            String key = routeId +"_"+timer.getLastTaskName();
            Long value = redisTemplate.opsForValue().decrement(key);
            log.info("end request taskName:{},seconds:{},key:{},value:{}",timer.getLastTaskName(),timer.getTotalTimeSeconds(),key,value);
            log.info("end --> {}",timer.prettyPrint());
        });
    }

    private ServerHttpResponseDecorator writeRsp(ServerWebExchange exchange, String routeId, String url) {
        //获取response的 返回数据
        ServerHttpResponse originalResponse = exchange.getResponse();
        DataBufferFactory bufferFactory = originalResponse.bufferFactory();
        ServerHttpResponseDecorator rsp = new ServerHttpResponseDecorator(originalResponse) {
            @Override
            public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
                if (getStatusCode().equals(HttpStatus.OK) && body instanceof Flux) {
                    Flux<? extends DataBuffer> fluxBody = Flux.from(body);
                    return super.writeWith(fluxBody.map(dataBuffer -> {
                        byte[] content = new byte[dataBuffer.readableByteCount()];
                        dataBuffer.read(content);
                        //释放掉内存
                        DataBufferUtils.release(dataBuffer);
                        //responseData就是下游系统返回的内容,可以查看修改
                        String responseData = new String(content, Charset.forName("UTF-8"));
                        //TODO 根据具体的 规则进行接口异步回调
                        asynNotice(responseData,routeId,url);
                        log.info("");
                        log.info("***********************************响应信息**********************************");
                        log.info("响应内容:{}", responseData);
                        log.info("****************************************************************************\n");
                        byte[] uppedContent = new String(content, Charset.forName("UTF-8")).getBytes();
                        return bufferFactory.wrap(uppedContent);
                    }));
                } else {
                    log.error("响应code异常:{}", getStatusCode());
                    asynNoticeError(getStatusCode(),routeId,url);
                }
                return super.writeWith(body);
            }
        };

        return rsp;
    }

    private void asynNoticeError(HttpStatus statusCode, String routeId, String url) {
        try{
            //TODO 规则实现
            //HttpUtil.doGet()
        }catch (Exception e){
            log.error(e.getMessage(),e);
        }
    }

    /**
     * 根据具体的 规则进行接口异步回调
     * @param responseData
     * @param routeId
     * @param url
     */
    private void asynNotice(String responseData, String routeId, String url) {
        try{
            //TODO 规则实现
            //HttpUtil.doGet()
        }catch (Exception e){
            log.error(e.getMessage(),e);
        }
    }

    /**
     * 按 路由 ips 选择一个 执行队列最小的
     * @param ips
     * @param routeId
     * @return
     */
    private String obtainRandomIp(List<String> ips,String routeId){
        log.info("obtainRandomIp ips:{}",JSON.toJSONString(ips));
        String key = null;
        ValueOperations opsValue = redisTemplate.opsForValue();
        Long start = 0L;
        String result = null;
        String lastKey = null;
        Random random = new Random();
        if(random.nextBoolean()){
            Collections.sort(ips);
        }else{
            //倒序
            Collections.reverse(ips);
        }
        for(String ip : ips){
            key = routeId +"_";
            key = key + ip ;
            log.info("key:{}",key);
            //按 key 加 1
            Long value = opsValue.increment(key);
            //按 key 减 1
            value = opsValue.decrement(key);
            log.info("start key:{},value:{}",key,value);
            value = Math.abs(value);
            log.info("end key:{},value:{}",key,value);
            KEYS.put(key,value);
            //取出 最小的值 start
            if(value <= start || start == 0){
                start = value;
                result = ip;
                lastKey = key;
            }
        }
        if(!StringUtils.isEmpty(lastKey)){
            //按 key 加 1
            Long value = opsValue.increment(lastKey);
            //TODO 如果 value 过大 则 自动启动新服务器
        }
        log.info("result:{}",result);
        return result;
    }
    /**
     * 通过 现有支持的ips 和正在执行中的 requestIps 获取一个可以执行的IP
     * 按 routId+ip 生成 key ,value 自动加减
     * 返回 ip
     * @param ips
     * @return
     */
    private String obtainRandomIp(List<String> ips,String routeId, boolean nextFlag) throws InterruptedException {
        // 获取所有的
        //ListOperations<String, String> listOps = redisTemplate.opsForList();
        List<String> requestIps = redisTemplate.opsForList().range(routeId+"_IPS_DOING", 0, -1);
        log.info("requestIps:{}", JSON.toJSON(requestIps));
        List<String> listTemp = new ArrayList<>(ips);
        if(requestIps != null && requestIps.size() > 0){
            listTemp.removeAll(requestIps);
        }
        if(listTemp.size() >0){
            Random random = new Random();
            return listTemp.get(random.nextInt(listTemp.size()));
        }
        //等于0 则说明 目前没有剩余资源 则等待1-3秒
        if(nextFlag){
            Thread.sleep(3*1000);
            return obtainRandomIp(ips,routeId,false);
        }
        //移出并获取列表的第一个元素， 如果列表没有元素会阻塞列表直到等待超时。
        //String ip = (String) redisTemplate.opsForList().leftPop(IPS_DOING, 3, TimeUnit.SECONDS);
        return null;
    }

    @Override
    public int getOrder() {
        return -1;
    }

    protected Mono<Void> forbidden(ServerWebExchange exchange) {
        return writeResponse(exchange,
                RspEntity.ofFail(String.valueOf(HttpStatus.FORBIDDEN.value()), HttpStatus.FORBIDDEN.getReasonPhrase()));
    }

    protected Mono<Void> writeResponse(ServerWebExchange exchange, RspEntity responseEntity) {
        // 设置response状态为ok
        exchange.getResponse().setStatusCode(HttpStatus.OK);

        // 响应
        if (responseEntity != null) {
            // 将traceId放入RspEntity对象中
            //responseEntity.setTraceId(easyMsApiAnalysisBean.getTraceId());
            String responseData = JSON.toJSONString(responseEntity);

            // 获取返回字符串的字节码，如果需要加密则返回的是加密后的字节码
            byte[] responseContent = responseData.getBytes(StandardCharsets.UTF_8);

            // 将byte放入DataBuffer中并返回
            String responseContentType = MediaType.APPLICATION_JSON_VALUE;
            ServerHttpResponse response = exchange.getResponse();
            response.getHeaders().add("Content-Type", responseContentType);
            response.getHeaders().setContentLength(responseContent.length);

            // 返回响应结果
            return response.writeWith(Mono.just(response.bufferFactory().wrap(responseContent)));
        }

        // 返回响应结果
        return Mono.empty();
    }


}
